package com.tech.process

import com.tech.common.{KafkaUtil, KuduUtil}
import com.tech.config.ApplicationConfig
import com.tech.process.EventProcess.{createDelayNode, createNode}
import com.tech.process.nodeProcess.{BehaviorNode, NextNode, TestNode, WaitNode}
import org.json4s.DefaultFormats
import org.json4s.jackson.JsonMethods

import scala.collection.mutable
import scala.util.Random
import scala.util.control.Breaks

object DynamicCustomerProcess {

  def dynamicCustomerProcess(value: String,
                             nextNodeMap: mutable.HashMap[String, NextNode],
                             behaviorNodeMap: mutable.HashMap[String, BehaviorNode],
                             testNodeMap: mutable.HashMap[String, TestNode],
                             waitNodeMap: mutable.HashMap[String, WaitNode]
                            ): Unit = {

    println("进入方法")


    implicit val f: DefaultFormats.type = org.json4s.DefaultFormats

    val map = JsonMethods.parse(value).extract[Map[String, Any]]

    val customer_id = map("customer_id").toString
    val activity_id = map("activity_id").toString
    val process_id = map("process_id").toString
    val member_id = map("member_id").toString

    println("nextNodeMap.contains(process_id):" + nextNodeMap.contains(process_id))

    //检查用户是否参与活动
    if (!KuduUtil.check(customer_id, activity_id) && nextNodeMap.contains(process_id)) {

      KuduUtil.insertCustomerActivity(customer_id, activity_id)
      pass(customer_id, activity_id, process_id, member_id, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap)

    }


  }

  def pass(customer_id: String,
           activity_id: String,
           process_id: String,
           member_id: String,
           nextNodeMap: mutable.HashMap[String, NextNode],
           behaviorNodeMap: mutable.HashMap[String, BehaviorNode],
           testNodeMap: mutable.HashMap[String, TestNode],
           waitNodeMap: mutable.HashMap[String, WaitNode]): Unit = {

    //    println("向kafka发送消息")
    //    KafkaUtil.send2KafkaAndWrite2Kudu(GlobalValue.customerNodeGroupTopic, GlobalValue.project, activity_id, process_id, customer_id, member_id)
    println("向kudu存储消息")
    KuduUtil.insertPassNodeMsg(customer_id, activity_id, process_id)
    passNode(customer_id, activity_id, process_id, member_id, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap)
  }

  def passNode(customer_id: String,
               activity_id: String,
               node: String,
               member_id: String,
               nextNodeMap: mutable.HashMap[String, NextNode],
               behaviorNodeMap: mutable.HashMap[String, BehaviorNode],
               testNodeMap: mutable.HashMap[String, TestNode],
               waitNodeMap: mutable.HashMap[String, WaitNode]): Unit = {

    /**
     * @param currentNode 当前节点
     */
    def toNextNode(currentNode: String): Unit = {
      val nextIds = nextNodeMap(currentNode).next_id

      if (nextIds != null)
        nextIds.split(",").foreach(
          next =>
            //100000000 开始
            //100000001 结束
            //100000002 用户
            //100000003 A/B Test
            //100000004 A/B Test Child
            //100000005 行为
            //100000006 奖励
            //100000007 等待
            //100000008 消息
            nextNodeMap(next).id_type match {

              case 100000001 =>
              case 100000002 =>
                //                KafkaUtil.send2KafkaAndWrite2Kudu(GlobalValue.customerNodeGroupTopic, GlobalValue.project, activity_id, next, customer_id, member_id)
                KuduUtil.insertPassNodeMsg(customer_id, activity_id, next)
                toNextNode(next)
              case 100000003 =>
                //                KafkaUtil.send2KafkaAndWrite2Kudu(GlobalValue.customerNodeGroupTopic, GlobalValue.project, activity_id, next, customer_id, member_id)
                KuduUtil.insertPassNodeMsg(customer_id, activity_id, next)
                val loop = new Breaks
                val random = new Random()
                val pro = random.nextInt(100)
                var tmp = 0
                loop.breakable {
                  nextNodeMap(next).next_id.split(",").foreach(
                    tmpNode => {
                      tmp += testNodeMap(tmpNode).probability
                      if (tmp > pro) {
                        println("customer_id:" + customer_id, pro)
                        //做random分流
                        //                        KafkaUtil.send2KafkaAndWrite2Kudu(GlobalValue.customerNodeGroupTopic, GlobalValue.project, activity_id, tmpNode, customer_id, member_id)
                        KuduUtil.insertPassNodeMsg(customer_id, activity_id, tmpNode)
                        toNextNode(tmpNode)
                        loop.break()
                      }
                    }
                  )
                }
              case 100000004 =>
                //                KafkaUtil.send2KafkaAndWrite2Kudu(GlobalValue.customerNodeGroupTopic, GlobalValue.project, activity_id, next, customer_id, member_id)
                KuduUtil.insertPassNodeMsg(customer_id, activity_id, next)
                //直接通过
                toNextNode(next)
              case 100000005 =>
                createDelayNode(customer_id, member_id, activity_id, next, behaviorNodeMap)
                createNode(customer_id, activity_id, next, behaviorNodeMap, System.currentTimeMillis())
              case 100000006 =>
                //发送消息到kafka
                println("奖励")
                KafkaUtil.send2Kafka(ApplicationConfig.CUSTOMER_NODE_GROUP_TOPIC, ApplicationConfig.PROJECT, activity_id, next, customer_id, member_id)
                KuduUtil.insertPassNodeMsg(customer_id, activity_id, next)
                toNextNode(next)
              case 100000007 =>
                //发送消息到延时事件表
                val action_time = if (waitNodeMap(next).wait_type == 100000000) {
                  System.currentTimeMillis() + (waitNodeMap(next).unit match {
                    case 100000001 => waitNodeMap(next).wait_duration.toLong * 1000 * 60 * 60 * 24 * 30
                    case 100000002 => waitNodeMap(next).wait_duration.toLong * 1000 * 60 * 60 * 24 * 7
                    case 100000003 => waitNodeMap(next).wait_duration.toLong * 1000 * 60 * 60 * 24
                    case 100000004 => waitNodeMap(next).wait_duration.toLong * 1000 * 60 * 60
                  })
                } else {
                  waitNodeMap(next).wait_date * 1000
                }
                KuduUtil.insertDelayNode(ApplicationConfig.PROJECT, customer_id, member_id, activity_id, next, action_time, "wait", is_group = "0")
              case 100000008 =>
                //发送消息到kafka
                println("消息")
                KafkaUtil.send2Kafka(ApplicationConfig.CUSTOMER_NODE_GROUP_TOPIC, ApplicationConfig.PROJECT, activity_id, next, customer_id, member_id)
                KuduUtil.insertPassNodeMsg(customer_id, activity_id, next)
                toNextNode(next)

            }
        )

    }

    toNextNode(node)

  }

}
